package com.czk.springboot.reactive.reactor.thread;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadDemo {
    public static void main(String[] args) {
        // 流的发布、中间操作默认使用当前线程
        Flux.range(0, 10)
                // 改变发布者的所在线程池在哪个线程池中执行
                .publishOn(Schedulers.immediate()) // 发布使用当前线程--main线程
                .log()
                .map(e -> e + 10)
               // .log()
                .subscribe(System.out::println);

        Flux.range(0, 10)
                .log()
                .map(e -> e + 10)
               // .log()
                // 改变订阅者所在线程池在哪个线程池中执行
                .subscribeOn(Schedulers.single())
                .subscribe(e-> System.out.println(Thread.currentThread().getName()+"===="+e));

        // 调度器
        Schedulers.boundedElastic(); // 有界，弹性调度，不是无界线程池，线程池中10*cpu核心个线程，队列默认100k
        Schedulers.immediate(); // 默认，无执行上下文，当前线程运行所有操作
        Schedulers.single(); // 使用固定的线程池

        Flux.range(0, 10)
                .publishOn(Schedulers.fromExecutor(new ThreadPoolExecutor(2,2,0, TimeUnit.MINUTES,new ArrayBlockingQueue<>(100),new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("czk的线程");
                        
                        return thread;
                    }
                },new ThreadPoolExecutor.DiscardOldestPolicy()))) // 发布使用当前线程--main线程
                .log()
                .map(e -> e + 10)
                .log()
                // 只要不指定线程，默认的是发布者的线程
                .subscribe(System.out::println);




    }
}
